package com.wyh.apitest.sink


import java.sql.{Connection, DriverManager, PreparedStatement}
import java.text.SimpleDateFormat
import java.util.Date

import com.wyh.apitest.source.CustomSource.MySensorSource
import com.wyh.apitest.source.SensorReading
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
import org.apache.flink.streaming.api.scala._

object JDBCSinkDemo {
  def main(args: Array[String]): Unit = {
    //定义流式处理环境
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)

    val sensorinfo: DataStream[String] = streamEnv.readTextFile("D:\\IdeaProjects\\FlinkStudyScala\\src\\main\\resources\\sensor.txt")


//    val stream: DataStream[SensorReading] = streamEnv.addSource(new MySensorSource)

    //转换成样例类类型
    val dataStream = sensorinfo.map(s => {
      val arr: Array[String] = s.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })


    dataStream.addSink(new MyJdbcFunction)
//    stream.addSink(new MyJdbcFunction)


    streamEnv.execute("JDBC Sink Demo")
  }

}

class MyJdbcFunction extends RichSinkFunction[SensorReading] {

  //定义连接、预编译sql语句
  var conn:Connection = _
  var insertStmt:PreparedStatement = _
  var updateStmt:PreparedStatement = _


  override def open(parameters: Configuration): Unit = {
    conn = DriverManager.getConnection("jdbc:mysql://192.168.205.100:3306/wyh_db", "root", "123456")
    insertStmt = conn.prepareStatement("insert into sensor_temp (id,temp,date_time) values(?,?,?)")
    updateStmt = conn.prepareStatement("update sensor_temp set temp = ? and date_time = ? where id = ?")
  }

  override def invoke(value: SensorReading): Unit = {
//    val timestamp = value.timestamp

    val df = new SimpleDateFormat()
    val timestemp = System.currentTimeMillis()
//    val date = new Date()
    val date_time = df.format(timestemp)

    //先执行更新操作，查到就更新
    updateStmt.setDouble(1,value.temperature)
    updateStmt.setString(2,date_time)
    updateStmt.setString(3,value.id)
    updateStmt.execute()

    //如果更新没有查到数据，那么就插入
    if(updateStmt.getUpdateCount == 0){
      insertStmt.setString(1,value.id)
      insertStmt.setDouble(2,value.temperature)
      insertStmt.setString(3,date_time)
      insertStmt.execute()
    }
  }

  override def close(): Unit = {
    insertStmt.close()
    updateStmt.close()
    conn.close()
  }

}